package com.atguigu.flink.chapter11.window;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * @Author lzc
 * @Date 2022/11/1 09:34
 */
public class Flink01_Window_Group {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<WaterSensor> stream = env.fromElements(
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s2", 1000L, 10),
            new WaterSensor("s1", 2000L, 20),
            new WaterSensor("s1", 3000L, 30),
            new WaterSensor("s1", 4000L, 40),
            new WaterSensor("s1", 5000L, 50),
            new WaterSensor("s1", 8000L, 50)
        ).assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((ws, ts) -> ws.getTs())
        );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts"), $("vc"), $("et").rowtime());
        
        // 定义一个滚动窗口: over 窗口长度  on 时间字段  as 给窗口起名字
//        GroupWindow win = Tumble.over(Expressions.lit(5).second()).on($("et")).as("w");
//        GroupWindow win = Slide.over( lit(5).second() ).every( lit(2).second() ).on( $("et") ).as("w");
        GroupWindow win = Session.withGap(lit(2).second()).on($("et")).as("w");
        table
            .window(win)
            .groupBy($("id"), $("w"))
            .select($("id"), $("w").start().as("stt"), $("w").end().as("edt"), $("vc").sum().as("vc_sum"))
            .execute()
            .print();
        
        
    }
}
/*
分组窗口(group window)
    keyBy-> window-> agg
    
    0-5  s1  10
    0-5  s2  100
    5-10 s1  20
    ...
    select w.start, w.end id, sum(vc) from t group by id, w
    


over 窗口


 */